/*
 * This example shows how to write a client that subscribes to a topic and does
 * not do anything other than handle the messages that are received.
 */
#include <iostream>
#include <assert.h>
#include <mosquitto.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <thread>
#include <atomic>
#include "zlog_app.h"
#include "config.h"
#include "app_main.h"
#include "mqtt_parse.h"

struct mosquitto *mosq;

static std::atomic<bool> s_stop;
static zlog_category_t *zlog_mqtt;
void publish_sensor_data(struct mosquitto *mosq, const char *topic, const string &json_string);
/* Callback called when the client receives a CONNACK message from the broker. */
void on_connect(struct mosquitto *mosq, void *obj, int reason_code)
{
    int rc;
    /* Print out the connection result. mosquitto_connack_string() produces an
     * appropriate string for MQTT v3.x clients, the equivalent for MQTT v5.0
     * clients is mosquitto_reason_string().
     */
    zlog_info(zlog_mqtt, "on_connect: %s", mosquitto_connack_string(reason_code));
    if (reason_code != 0)
    {
        /* If the connection fails for any reason, we don't want to keep on
         * retrying in this example, so disconnect. Without this, the client
         * will attempt to reconnect. */
        mosquitto_disconnect(mosq);
    }

    /* Making subscriptions in the on_connect() callback means that if the
     * connection drops and is automatically resumed by the client, then the
     * subscriptions will be recreated when the client reconnects. */
    // char sub_topic[128];
    // snprintf(sub_topic, sizeof(sub_topic), "Ser/+/+/%s", get_app_config_ptr()->get_dev_config()->dev_ip.c_str());
    const char *sub_topic = "Dev/#";
    rc = mosquitto_subscribe(mosq, NULL, sub_topic, 1);
    zlog_info(zlog_mqtt, "mosquitto_subscribe topic: %s", sub_topic);

    if (rc != MOSQ_ERR_SUCCESS && get_app_config_ptr()->cfg.auto_get_threshold)
    {
        zlog_warn(zlog_mqtt, "mosquitto_subscribe topic: %s", mosquitto_strerror(rc));
        /* We might as well disconnect if we were unable to subscribe */
        mosquitto_disconnect(mosq);
    }
    // 获取阈值参数
    get_send_down_queue_ptr()->push(EVENT::GET_THRESHOLD_PARAM);
}

/* Callback called when the broker sends a SUBACK in response to a SUBSCRIBE. */
void on_subscribe(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
    int i;
    bool have_subscription = false;

    /* In this example we only subscribe to a single topic at once, but a
     * SUBSCRIBE can contain many topics at once, so this is one way to check
     * them all. */
    /* 遍历所有订阅主题的QoS等级 */
    for (i = 0; i < qos_count; i++)
    {
        zlog_info(zlog_mqtt, "on_subscribe: %d:granted qos = %d", i, granted_qos[i]);
        if (granted_qos[i] <= 2)
        {
            have_subscription = true;
        }
    }
    /* 如果所有订阅都被拒绝，则断开连接 */
    if (have_subscription == false)
    {
        /* The broker rejected all of our subscriptions, we know we only sent
         * the one SUBSCRIBE, so there is no point remaining connected. */
        zlog_warn(zlog_mqtt, "Error: All subscriptions rejected.");
        mosquitto_disconnect(mosq);
    }
}
void on_publish(struct mosquitto *mosq, void *obj, int mid)
{
    zlog_info(zlog_mqtt, "Message with mid %d has been published.", mid);
}
/* Callback called when the client receives a message. */
void on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
    /* This blindly prints the payload, but the payload can be anything so take care. */
    zlog_info(zlog_mqtt, "msg {\n\ttopic:%s\n\tqso: %d\n\tpayload= %s\n}", msg->topic, msg->qos, (char *)msg->payload);

    std::vector<std::string> parts = split_topic((char *)msg->topic);
    if (parts.size() != 2)
    {
        zlog_error(zlog_mqtt, "Invalid heartbeat message");
        return;
    }
    /*================================Heart   =======================================*/
    if ("Heart" == parts[1])
    {
        struct DeviceHeartbeat heart;
        if (parse_heart_msg(heart, (char *)msg->payload))
        {
            zlog_info(zlog_mqtt, "ProtocolID: %s, DevType: %d, DevIP: %s, HeartTime: %s, HeartInterval: %d, TestNum: %d, AlarmNum: %d",
                      heart.ProtocolID.c_str(), heart.DevType, heart.DevIP.c_str(), heart.HeartTime.c_str(), heart.HeartInterval, heart.TestNum, heart.AlarmNum);

            std::string json = build_heartbeat_json(heart);
            zlog_info(zlog_mqtt, "repay json: %s", json.c_str());
            publish_sensor_data(mosq, "Ser/Heart", json);
            // 更新数据
            fillDevDataFromHeartbeat(heart, *get_dev_data_ptr());
            get_send_up_queue_ptr()->push(EVENT::HEART_BEAT);
        }
        else
        {
            zlog_warn(zlog_mqtt, "parse_heart_msg failed");
        }
    }
    /*================================ AlarmThresholdSet   =======================================*/
    else if ("AlarmThresholdSet" == parts[1])
    {
        AlarmThresholdConfig _dev_data;
        if (load_alarm_threshold_from_json(_dev_data, (char *)msg->payload))
        {
            zlog_info(zlog_mqtt, "load_alarm_threshold_from_json success");
            get_app_status_flag_ptr()->get_threshold_config_ok = 1; // 获取阈值成功
                                                                    // 更新数据
            fillDevDataFromAlarmThresholdConfig(_dev_data, *get_dev_data_ptr());
            // 发送更新后的阈值配置到设备
            get_send_up_queue_ptr()->push(EVENT::GET_DEV_CONFIG);
        }
        else
        {
            zlog_warn(zlog_mqtt, "load_alarm_threshold_from_json failed");
        }
    }
    /*================================ RspData   RspDataResult  =======================================*/
    else if ("RspData" == parts[1] || "RspDataResult" == parts[1]) // 实时数据，测量结果
    {
        rsp_data_info info;
        if (parseDetectorData((char *)msg->payload, info))
        {
            zlog_info(zlog_mqtt, "parseDetectorData success");
            // 更新数据
            fillDevDataFromRspDataInfo(info, *get_dev_data_ptr());
            zlogDevData(zlog_mqtt, *get_dev_data_ptr(), "RspData");
            // udp
            if (get_app_config_ptr()->cfg.auto_push_data)
            {
                // 发送不同的数据
                get_send_up_queue_ptr()->push(EVENT::RSP_DATA_BASE_STATUS);
                get_send_up_queue_ptr()->push(EVENT::RSP_DATA_BASE_ALPHA);
                get_send_up_queue_ptr()->push(EVENT::RSP_DATA_BASE_BETA);
                get_send_up_queue_ptr()->push(EVENT::RSP_DATA_BASE_GAMA);
                get_send_up_queue_ptr()->push(EVENT::RSP_DATA_BASE_OTHERS);
            }
            // modbus rtu
            if (DeviceStatus::FRONT_MEASUREMENT_RESULT == static_cast<DeviceStatus>(info.reg_status) ||
                DeviceStatus::BACK_MEASUREMENT_RESULT == static_cast<DeviceStatus>(info.reg_status))
            {
                get_send_modbusrtu_queue_ptr()->push(EVENT::RSP_DATA_BASE_STATUS);
                get_send_modbusrtu_queue_ptr()->push(EVENT::RSP_DATA_BASE_ALPHA);
                get_send_modbusrtu_queue_ptr()->push(EVENT::RSP_DATA_BASE_BETA);
                get_send_modbusrtu_queue_ptr()->push(EVENT::RSP_DATA_BASE_GAMA);
                get_send_modbusrtu_queue_ptr()->push(EVENT::RSP_DATA_BASE_OTHERS);
            }

            // 回复发送数据
            std::string json = build_get_rsp_num_json(info.RSPCount);
            publish_sensor_data(mosq, "Ser/RspData", json);
        }
        else if ("SetTime" == parts[1])
        {
            zlog_info(zlog_mqtt, "SetTime is seccess");
        }
        else
        {
            zlog_warn(zlog_mqtt, "parseDetectorData failed");
        }
    }
    else
    {
        zlog_warn(zlog_mqtt, "not topic match");
    }
}

/* This function pretends to read some data from a sensor and publish it.*/
void publish_sensor_data(struct mosquitto *mosq, const char *topic, const string &json_string)
{
    char payload[20];
    int temp;
    int rc;

    /* Print it to a string for easy human reading - payload format is highly
     * application dependent. */
    snprintf(payload, sizeof(payload), "%d", temp);

    /* Publish the message
     * mosq - our client instance
     * *mid = NULL - we don't want to know what the message id for this message is
     * topic = "example/temperature" - the topic on which this message will be published
     * payloadlen = strlen(payload) - the length of our payload in bytes
     * payload - the actual payload
     * qos = 2 - publish with QoS 2 for this example
     * retain = false - do not use the retained message feature for this message
     */
    zlog_info(zlog_mqtt, "MQTT SEND EVENT topic: %s, json_string: %s", topic, json_string.c_str());
    rc = mosquitto_publish(mosq, NULL, topic, strlen(json_string.c_str()), json_string.c_str(), 2, false);
    if (rc != MOSQ_ERR_SUCCESS)
    {
        zlog_warn(zlog_mqtt, "Error publishing: %s", mosquitto_strerror(rc));
    }
}
void set_mqtt_stop_flag(bool stop)
{
    s_stop = stop;
}
/**
 * @brief 发送设备事件到MQTT服务器
 *
 */
static void mqtt_send_dev_event(void)
{
    EVENT event;
    std::string json_str;
    if (get_send_down_queue_ptr()->try_pop(event))
    {
        zlog_info(zlog_mqtt, "MQTT RECV EVENT event: %d", event);
        switch (event)
        {
        case EVENT::SYNC_RTC_TIMESTAMP:
            json_str = build_time_sync_json();
            publish_sensor_data(mosq, "Ser/SetTime", json_str);

            break;
        case EVENT::GET_THRESHOLD_PARAM:
            if (get_app_config_ptr()->cfg.auto_get_threshold)
            {
                json_str = build_get_threshold_json();
                publish_sensor_data(mosq, "Ser/AlarmThresholdSet", json_str);
            }

            break;
        default:
            break;
        }
    }
}
int mqtt_init(void)
{
    int rc;

    /* Required before calling other mosquitto functions */
    mosquitto_lib_init();

    /* Create a new client instance.
     * id = NULL -> ask the broker to generate a client id for us
     * clean session = true -> the broker should remove old sessions when we connect
     * obj = NULL -> we aren't passing any of our private data for callbacks
     */
    zlog_mqtt = zlog_get_category(ZLOG_CAG_MQTT);
    assert(zlog_mqtt != NULL);
    mosq = mosquitto_new(NULL, true, NULL);
    if (mosq == NULL)
    {
        zlog_warn(zlog_mqtt, "Error: Out of memory.\n");
        return 1;
    }
    else
    {
        zlog_info(zlog_mqtt, "mosquitto_new() success");
    }

    /* Configure callbacks. This should be done before connecting ideally. */
    zlog_info(zlog_mqtt, "Configure callbacks 1.on_connect 2.on_subscribe 3.on_message 4.on_publish");
    mosquitto_connect_callback_set(mosq, on_connect);
    mosquitto_subscribe_callback_set(mosq, on_subscribe);
    mosquitto_message_callback_set(mosq, on_message);
    mosquitto_publish_callback_set(mosq, on_publish);

    /* Connect to test.mosquitto.org on port 1883, with a keepalive of 60 seconds.
     * This call makes the socket connection only, it does not complete the MQTT
     * CONNECT/CONNACK flow, you should use mosquitto_loop_start() or
     * mosquitto_loop_forever() for processing net traffic. */
    rc = mosquitto_username_pw_set(mosq, "admin", "admin");
    if (rc != MOSQ_ERR_SUCCESS)
    {
        zlog_warn(zlog_mqtt, "mosquitto_username_pw_set Error: %s", mosquitto_strerror(rc));
    }
    else
    {
        zlog_info(zlog_mqtt, "mosquitto_username_pw_set Success");
    }

    rc = mosquitto_connect(mosq, "127.0.0.1", 1883, 600);
    if (rc != MOSQ_ERR_SUCCESS)
    {
        mosquitto_destroy(mosq);
        zlog_warn(zlog_mqtt, "mosquitto_connect Error: %s", mosquitto_strerror(rc));
        return 1;
    }
    else
    {
        zlog_info(zlog_mqtt, "mosquitto_connect Success");
    }

    /* Run the network loop in a blocking call. The only thing we do in this
     * example is to print incoming messages, so a blocking call here is fine.
     *
     * This call will continue forever, carrying automatic reconnections if
     * necessary, until the user calls mosquitto_disconnect().
     */

    s_stop = false;

    // static std::thread mqtt_send([]()
    //                              {

    // while(!s_stop){
    // 	std::this_thread::sleep_for(std::chrono::milliseconds(100));
    // 	printf("mosquitto_loop_forever\n");
    // 		publish_sensor_data(mosq); } });

    // mosquitto_loop_forever(mosq, -1, 1);

    // // 等待线程结束,否则段错误
    // mqtt_send.join();

    while (!s_stop)
    {
        mosquitto_loop(mosq, 10, 2); // 每次最多处理 1 个网络包，超时 10ms

        // 在这里你可以插入其他业务逻辑或定时发布
        static int counter = 0;
        if (++counter % 500 == 0 &&
            0 == get_app_status_flag_ptr()->get_threshold_config_ok)
        {
            // 获取阈值参数
            if (get_app_config_ptr()->cfg.auto_get_threshold)
            {
                get_send_down_queue_ptr()->push(EVENT::GET_THRESHOLD_PARAM);
            }
        }
        // 处理mqtt设备事件

        mqtt_send_dev_event();
    }

    // 清理资源
    mosquitto_disconnect(mosq);
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
    cout << "mosquitto_loop_forever end" << endl;
    return 0;
}
